package collecter

import (
	kafka "multitude-kafka-exporter/kafka_exporter"
	"net/http"
	"strings"

	"github.com/golang/glog"
	"github.com/prometheus/client_golang/prometheus/promhttp"
)

type KafKaExporter struct {
	Mux                 *http.ServeMux
	MetricsPath         string
	ScrapePate          string
	DefautlTopicWorkers int
}

func (e *KafKaExporter) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	e.Mux.ServeHTTP(w, r)
}

func (e *KafKaExporter) HealthHandler(w http.ResponseWriter, r *http.Request) {
	_, _ = w.Write([]byte(`ok`))
}

func (e *KafKaExporter) IndexHandler(w http.ResponseWriter, r *http.Request) {
	_, _ = w.Write([]byte(`<html>
<head><title>Kafka Exporter </title></head>
<body>
<h1>Kafka Exporter</h1>
<p><a href='` + e.MetricsPath + `'>Metrics</a></p>
</body>
</html>
`))
}

func (e *KafKaExporter) ScrapeHandler(w http.ResponseWriter, r *http.Request) {
	target := r.URL.Query().Get("target")
	if target == "" {
		http.Error(w, "'target' parameter must be specified", http.StatusBadRequest)
		return
	}

	topicFilter := r.URL.Query().Get("topic_filter")
	if len(topicFilter) == 0 {
		topicFilter = ".*"
	}

	groupFilter := r.URL.Query().Get("group_filter")
	if len(groupFilter) == 0 {
		groupFilter = ".*"
	}

	zookeeper := r.URL.Query().Get("zookeeper_iplist")
	if len(zookeeper) == 0 {
		zookeeperlist := make([]string, 0)
		for _, host := range strings.Split(target, ",") {
			ip := strings.Split(host, ":")
			zookeeperip := strings.Join(
				[]string{
					ip[0],
					"2181",
				}, ":",
			)
			zookeeperlist = append(zookeeperlist, zookeeperip)
		}
		zookeeper = strings.Join(zookeeperlist, ",")

	}
	labels := r.URL.Query().Get("labels")

	kafkaVersion := r.URL.Query().Get("kafka_version")

	exporter, err := kafka.NewKafkaCollecter(target, kafkaVersion, zookeeper, labels, topicFilter, groupFilter, e.DefautlTopicWorkers)
	if err != nil {
		http.Error(w, "NewKafkaExporter() err: err", http.StatusBadRequest)
		return
	}
	defer exporter.Client.Close()
	glog.Info("exporter return ok")
	promhttp.HandlerFor(
		exporter.Registry, promhttp.HandlerOpts{ErrorHandling: promhttp.ContinueOnError},
	).ServeHTTP(w, r)
}
